package com.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gmall.realtime.common.base.BaseApp;
import com.gmall.realtime.common.bean.CartAddUuBean;
import com.gmall.realtime.common.constant.Constant;
import com.gmall.realtime.common.function.DorisMapFunction;
import com.gmall.realtime.common.util.DateFormatUtil;
import com.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsTradeCartAddUuWindow extends BaseApp {
    /**
     * Flink 实时任务启动主入口方法
     *
     * @param args 命令行参数（本实现中未使用）
     *
     * 方法逻辑说明：
     * 1. 创建当前类的实例并启动 Flink 实时处理任务
     * 2. 参数说明：
     *    - 10026: 应用运行的端口号
     *    - 4: Flink 任务的并行度设置
     *    - "dws_trade_cart_add_uu_window": Doris 目标表名称前缀
     *    - Constant.TOPIC_DWD_TRADE_CART_ADD: Kafka 数据源对应的主题名称
     * 3. 继承自 BaseApp 的 start 方法实现了 Flink 作业的基础配置和任务提交
     *
     * @throws Exception 可能抛出的运行时异常（如 Flink 任务提交失败等）
     */
    public static void main(String[] args) {
        new DwsTradeCartAddUuWindow().start(
                10026,
                1,
                "dws_trade_cart_add_uu_window",
                Constant.TOPIC_DWD_TRADE_CART_ADD
        );
    }

    /**
     * 处理 Kafka 数据源的主方法，实现购物车加购独立用户统计的核心流程
     *
     * @param env        Flink 流执行环境
     * @param kafkaSource 从 Kafka 读取的原始数据流
     *
     * 方法处理流程：
     * 1. 数据清洗：过滤无效数据并转换数据格式
     * 2. 添加水位线：处理事件时间语义和乱序事件
     * 3. 按用户ID分组：使用 KeyBy 进行用户维度分组
     * 4. 窗口聚合：统计10秒窗口内的独立用户加购数
     * 5. 结果输出：将聚合结果写入 Doris 数据库
     */
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaSource) {
        // 数据清洗阶段：过滤无效数据并转换JSON格式
        SingleOutputStreamOperator<JSONObject> etlStream = etl(kafkaSource);

        // 添加水位线阶段：设置3秒的允许延迟时间
        SingleOutputStreamOperator<JSONObject> watermarksStream = getWatermarksStream(etlStream);

        // 用户分组阶段：按user_id分组并处理用户首次加购行为
        SingleOutputStreamOperator<CartAddUuBean> KeyByStream = getKeyByStream(watermarksStream);

        // 窗口聚合阶段：10秒滚动窗口统计独立用户数
        SingleOutputStreamOperator<CartAddUuBean> reduceStream = getReduceStream(KeyByStream);

        // 结果输出阶段：将聚合结果映射为Doris格式并写入
        reduceStream
                .map(new DorisMapFunction<>())
                .sinkTo(
                        FlinkSinkUtil.getDorisSink(
                                Constant.DORIS_DATABASE + "." + Constant.DWS_TRADE_CART_ADD_UU_WINDOW,
                                Constant.DWS_TRADE_CART_ADD_UU_WINDOW
                        )
                );
    }


    /**
     * 对按用户分组后的数据流进行窗口聚合，统计10秒窗口内的独立用户加购数
     *
     * @param KeyByStream 已按用户ID分组并处理过的数据流，包含用户加购行为信息
     * @return SingleOutputStreamOperator<CartAddUuBean> 返回聚合后的数据流，包含窗口统计结果
     *
     * 方法处理逻辑：
     * 1. 使用10秒的滚动事件时间窗口进行窗口划分
     * 2. 在窗口内对用户加购数进行累加聚合（Reduce阶段）
     * 3. 处理窗口结果时添加窗口起止时间和当前日期信息（Process阶段）
     */
    private SingleOutputStreamOperator<CartAddUuBean> getReduceStream(SingleOutputStreamOperator<CartAddUuBean> KeyByStream) {
        // 设置10秒滚动窗口，基于事件时间
        return KeyByStream.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)))
                // Reduce阶段：累加窗口内所有用户的加购数
                .reduce(new ReduceFunction<CartAddUuBean>() {
                            @Override
                            public CartAddUuBean reduce(CartAddUuBean value1, CartAddUuBean value2) throws Exception {
                                value1.setCartAddUuCt(value1.getCartAddUuCt() + value2.getCartAddUuCt());
                                return value1;
                            }
                        },
                        // Process阶段：处理窗口结果，添加时间信息
                        new ProcessAllWindowFunction<CartAddUuBean, CartAddUuBean, TimeWindow>() {
                            @Override
                            public void process(ProcessAllWindowFunction<CartAddUuBean, CartAddUuBean, TimeWindow>.Context context,
                                                Iterable<CartAddUuBean> iterable,
                                                Collector<CartAddUuBean> collector) throws Exception {
                                // 获取当前窗口的时间范围
                                TimeWindow window = context.window();
                                long start = window.getStart();
                                long end = window.getEnd();

                                // 遍历窗口结果，设置时间字段
                                for (CartAddUuBean cartAddUuBean : iterable) {
                                    // 设置窗口开始时间（格式化为日期时间字符串）
                                    cartAddUuBean.setStt(DateFormatUtil.tsToDateTime(start));
                                    // 设置窗口结束时间（格式化为日期时间字符串）
                                    cartAddUuBean.setEdt(DateFormatUtil.tsToDateTime(end));
                                    // 设置当前处理日期（格式化为日期字符串）
                                    cartAddUuBean.setCurDate(DateFormatUtil.tsToDate(System.currentTimeMillis()));
                                    collector.collect(cartAddUuBean);
                                }
                            }
                        });
    }


    /**
     * 对已添加水位线的数据流按用户ID分组，并处理每个用户的首次加购行为
     *
     * @param watermarksStream 已添加水位线的数据流，包含用户加购行为JSON数据
     * @return SingleOutputStreamOperator<CartAddUuBean> 返回处理后的数据流，包含用户加购统计信息
     *
     * 方法处理逻辑：
     * 1. 按user_id进行分组，确保相同用户的数据进入同一个处理函数
     * 2. 使用KeyedProcessFunction处理每个用户的加购行为
     * 3. 通过状态管理确保每个用户每天只统计一次加购行为
     */
    private SingleOutputStreamOperator<CartAddUuBean> getKeyByStream(SingleOutputStreamOperator<JSONObject> watermarksStream) {
        // 按用户ID分组并处理每个用户的加购行为
        SingleOutputStreamOperator<CartAddUuBean> KeyByStream = watermarksStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getString("user_id");
            }
        }).process(new KeyedProcessFunction<String, JSONObject, CartAddUuBean>() {
            private ValueState<String> userAddState;  // 用于存储用户最后加购日期的状态变量

            @Override
            public void open(Configuration parameters) throws Exception {
                // 初始化状态描述符，设置TTL为1天（避免状态无限增长）
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("user_add", String.class);
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());
                userAddState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, CartAddUuBean>.Context context, Collector<CartAddUuBean> collector) throws Exception {
                // 获取事件时间戳并转换为日期格式
                Long ts = jsonObject.getLong("ts");
                String curDt = DateFormatUtil.tsToDate(ts);
                String value = userAddState.value();
                long cartAddUuCt = 0L;

                // 检查用户当天是否已有加购记录，若无则计数+1并更新状态
                if (!curDt.equals(value)) {
                    cartAddUuCt = 1L;
                    userAddState.update(curDt);
                }

                // 只输出有加购行为的记录（cartAddUuCt=1）
                if (cartAddUuCt != 0) {
                    collector.collect(CartAddUuBean
                            .builder()
                            .cartAddUuCt(cartAddUuCt)
                            .build());
                }
            }
        });
        return KeyByStream;
    }


    /**
     * 为数据流添加水位线和事件时间戳
     *
     * @param etlStream 经过ETL处理后的数据流，包含JSON格式的用户加购行为数据
     * @return SingleOutputStreamOperator<JSONObject> 返回添加了水位线和时间戳的数据流
     *
     * 方法处理逻辑：
     * 1. 使用有界无序水位线策略，允许3秒的乱序延迟
     * 2. 从JSON对象中提取ts字段作为事件时间戳
     * 3. 水位线生成基于事件时间戳，用于后续的窗口计算
     */
    private SingleOutputStreamOperator<JSONObject> getWatermarksStream(SingleOutputStreamOperator<JSONObject> etlStream) {
        return etlStream.assignTimestampsAndWatermarks(
                // 设置允许3秒的乱序延迟
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        // 从JSON对象中提取时间戳字段
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObject, long l) {
                                return jsonObject.getLong("ts");
                            }
                        }));
    }


    /**
     * 对 Kafka 原始数据进行 ETL 处理，过滤无效数据并转换时间戳格式
     *
     * @param kafkaSource 从 Kafka 读取的原始字符串数据流
     * @return SingleOutputStreamOperator<JSONObject> 返回处理后的 JSON 数据流
     *
     * 方法处理逻辑：
     * 1. 将字符串解析为 JSON 对象
     * 2. 检查必须字段（时间戳和用户ID）是否存在
     * 3. 将时间戳从秒级转换为毫秒级（Flink 标准时间单位）
     * 4. 输出有效的 JSON 数据对象
     */
    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> kafkaSource) {
        return kafkaSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                // 将 Kafka 消息字符串解析为 JSON 对象
                JSONObject jsonObject = JSON.parseObject(s);

                // 提取关键字段：时间戳和用户ID
                Long ts = jsonObject.getLong("ts");
                String userId = jsonObject.getString("user_id");

                // 校验字段有效性：两个字段都不能为null
                if (ts != null && userId != null) {
                    // 处理时间戳为毫秒值（Flink 内部使用毫秒时间戳）
                    jsonObject.put("ts", ts * 1000);
                    // 收集有效数据到下游
                    collector.collect(jsonObject);
                }
            }
        });
    }

}